#include "zmqproxy.h"
#include <cstring>
#include <stdlib.h>
#include <stdio.h>
#include <iostream>
#include <cassert>
#include "zmq.h"
#include "json/json.h"


void* AbsZmqProxy::context = zmq_ctx_new();
std::atomic<int> AbsZmqProxy::context_in_used;

AbsZmqProxy::AbsZmqProxy(int SOCKET_TYPE)
{
    socket = zmq_socket(context, SOCKET_TYPE);
    ++context_in_used;
    // 多播默认ttl为1，表示仅在本地网络中广播，不能经过多个网络设备
    // 防止经过多个设备后，无法接收到网络消息
    if(SOCKET_TYPE == ZMQ_PUB)
    {
        int ttl = 10;
        zmq_setsockopt(socket, ZMQ_MULTICAST_HOPS, &ttl, sizeof(ttl));
    }
}

AbsZmqProxy::~AbsZmqProxy()
{
    --context_in_used;
    printf("context_in_used: %d\n", (int)context_in_used);
    zmq_close(socket);
    if(context_in_used == 0)
    {
        zmq_ctx_destroy(context);
        printf("[AbsZmqProxy] zmq context Released!\n");
    }
}

void AbsZmqProxy::ReleaseSocket()
{
    SetLinger();
    zmq_close(socket);
    socket = nullptr;
    --context_in_used;
}

void AbsZmqProxy::CreateNewSocket(int SOCKET_TYPE)
{
    socket = zmq_socket(context, SOCKET_TYPE);
    ++context_in_used;
}

int AbsZmqProxy::SetSNDHWD(int hwd)
{
    return zmq_setsockopt(socket, ZMQ_SNDHWM, &hwd, sizeof(hwd));
}

int AbsZmqProxy::SetRCVHWD(int hwd)
{
    return zmq_setsockopt(socket, ZMQ_RCVHWM, &hwd, sizeof(hwd));
}

int AbsZmqProxy::SetRCVBUF(int rcv_buff)
{
    return zmq_setsockopt(socket, ZMQ_RCVBUF, &rcv_buff, sizeof(rcv_buff));
}

int AbsZmqProxy::SetSNDBUF(int snd_buff)
{
    return zmq_setsockopt(socket, ZMQ_RCVBUF, &snd_buff, sizeof(snd_buff));
}

int AbsZmqProxy::SetRATE(int rate)
{
    return zmq_setsockopt(socket, ZMQ_RATE, &rate, sizeof(rate));
}

int AbsZmqProxy::SetRECOVERYIVL(int ivl)
{
    return zmq_setsockopt(socket, ZMQ_RECOVERY_IVL, &ivl, sizeof(ivl));
}

int AbsZmqProxy::KeepLastMsg()
{
    int val = 1;
    return zmq_setsockopt(socket, ZMQ_CONFLATE, &val, sizeof(val));;
}

int AbsZmqProxy::recv(std::string &sMsg, int flags)
{
    return recv(socket, sMsg, flags);
}

int AbsZmqProxy::send(const std::string& sMsg, int flags)
{
    return send(socket, sMsg, flags);
}

int AbsZmqProxy::recv(void* socket, std::string &sMsg, int flags)
{
    // TODO: socket 异常需要重启
    zmq_msg_t zMsg;
    zmq_msg_init(&zMsg);
    int size = zmq_msg_recv(&zMsg, socket, flags);
    if(size == -1) return -1;
    sMsg.assign((const char *)zmq_msg_data(&zMsg), size);
    // printf("[pgmsock] receive msg: \n%s\n", msg.c_str());
    zmq_msg_close(&zMsg);
    return sMsg.length();
}

int AbsZmqProxy::send(void* socket, const std::string &sMsg, int flags)
{
    zmq_msg_t zMsg;
    zmq_msg_init_size(&zMsg, sMsg.length());
    memcpy(zmq_msg_data(&zMsg), sMsg.c_str(), sMsg.length());
    int rc = zmq_msg_send(&zMsg, socket, flags);
    zmq_msg_close(&zMsg);
    return rc;
}

void AbsZmqProxy::SetLinger()
{
    int linger = 0;
    zmq_setsockopt(socket, ZMQ_LINGER, &linger, 4);
}

ZmqPub::ZmqPub(int port, const std::string& protocal): AbsZmqProxy(ZMQ_PUB), protocal(protocal)
{
    char addr[64];
    sprintf(addr, "%s://*:%d", protocal.data(), port);
    int rc = zmq_bind(socket, addr);
    if (rc != 0)
    {
        std::cout << zmq_strerror(errno) << std::endl;
        std::cout << "[zmqtcppub] Create Publisher Failed!" << std::endl;
    }
}

ZmqPub::~ZmqPub()
{
    zmq_close(socket);
}

int ZmqPub::pub(const std::string& channel, const std::string& sMsg, int flags)
{
    std::string str("[" + channel + "]" + sMsg);

    zmq_msg_t zMsg;
    zmq_msg_init_size(&zMsg, str.length());
    memcpy(zmq_msg_data(&zMsg), str.c_str(), str.length());

    int rc = zmq_msg_send(&zMsg, socket, 0);
    if(rc == -1)
    {
        std::cout << "[zmqtcppub] pub failed!" << std::endl;
        std::cerr << "[zmqtcppub] " << zmq_strerror(zmq_errno()) << std::endl;
    }

    zmq_msg_close(&zMsg);

    return rc;
}

int ZmqPub::unpacking_pub(const std::string &channel, const std::string &sMsg)
{
    // 设置拆分的大小
    std::string data("[" + channel + "]" + sMsg);
    int rc = 0;
    int offset = 0;

    while(data.begin() + offset < data.end())
    {
        int size = std::min(MAX_CHUNK, (int)data.length() - offset);
        // 创建消息块
        zmq_msg_t message;
        zmq_msg_init_size(&message, size);
        memcpy(zmq_msg_data(&message), data.c_str() + offset, size);

        offset += MAX_CHUNK;
        rc = zmq_msg_send(&message, socket, data.begin() + offset < data.end() ? ZMQ_SNDMORE : 0);
        if(rc == -1)
        {
            std::cout << "[zmqtcppub] pub failed!" << std::endl;
            std::cerr << "[zmqtcppub] " << zmq_strerror(zmq_errno()) << std::endl;
            break;
        }

    }

    return rc;
}

ZmqRep::ZmqRep(int port, const std::string& protocal): AbsZmqProxy(ZMQ_REP), protocal(protocal)
{
    char addr[64];
    sprintf(addr, "%s://*:%d", protocal.data(), port);
    int rc = zmq_bind(socket, addr);
    if (rc != 0)
    {
        std::cout << zmq_strerror(errno) << std::endl;
        std::cout << "[zmqtcppub] Create Publisher Failed!" << std::endl;
    }
}

ZmqRep::~ZmqRep()
{
    zmq_close(socket);
}

int ZmqRep::recv(std::string &sMsg)
{
    return AbsZmqProxy::recv(sMsg);
}


int ZmqRep::response(std::string &&sMsg)
{
    return AbsZmqProxy::send(sMsg);
}

ZmqPair::ZmqPair() : AbsZmqProxy(ZMQ_PAIR)
{}

ZmqPair::ZmqPair(int port):AbsZmqProxy(ZMQ_PAIR), port(port) 
{}

ZmqPair::~ZmqPair()
{
    printf("[ZmqPair] Released Port: %d!\n", port);
}

int ZmqPair::CreateRecevier()
{
    char addr[64] {0};
    sprintf(addr, "tcp://*:%d", port);
    printf("[ZmqPair] Bind Addr: %s\n", addr);
    int rc = zmq_bind(socket, addr);
    if(rc != 0)
    {
        printf("%s\n", zmq_strerror(zmq_errno()));
    }
    assert(rc == 0);
    return rc;
}

int ZmqPair::recv(std::string &sMsg, int flags)
{
    return AbsZmqProxy::recv(sMsg, flags);
}

int ZmqPair::send(const std::string& ip, const std::string& sMsg, int flags)
{
    char addr[64] {0};
    sprintf(addr, "tcp://%s:%d", ip.data(), port);
    sender = zmq_socket(context, ZMQ_PAIR);
    int rc = zmq_connect(sender, addr);
    if(rc != 0)
    {
        printf("%s\n", zmq_strerror(zmq_errno()));
    }
    printf("[ZmqPair] send to %s msg len %d\n", addr, sMsg.length());
    assert(rc == 0);
    int iRet = AbsZmqProxy::send(sender, sMsg, flags);
    if(iRet < 0)
    {
        printf("zmq err: %s\n", zmq_strerror(zmq_errno()));
    }
    zmq_close(sender);
    return iRet;
}


// int main(int argc, char const *argv[])
// {
//     ZmqPair pair1("5556");
//     pair1.send("127.0.0.1", "1111");
//     printf("finished\n");
//     ZmqPair pair2("5556");
//     std::string sMsg;
//     pair2.recv(sMsg);
//     std::cout << sMsg << std::endl;
//     return 0;
// }


